#include <basic/errcode.h>
#include <basic/msgq.h>

static int64_t _begin(int timeout_ms, struct timespec *ts)
{/*{{{*/
    struct timeval tv;
    gettimeofday(&tv, NULL);
    int64_t begin=tv.tv_sec*1000LL+tv.tv_usec/1000;
#if __WORDSIZE < 64
    ts->tv_sec=tv.tv_sec+timeout_ms/1000+1;
    ts->tv_nsec=0;
#else
    ts->tv_sec=tv.tv_sec + timeout_ms/1000;
    ts->tv_nsec=(tv.tv_usec + timeout_ms%1000*1000L) * 1000L;
    if(ts->tv_nsec > 1000000000L)
    {
        ts->tv_sec+=1;
        ts->tv_nsec-=1000000000L;
    }
#endif
    return(begin);
}/*}}}*/

static int _left(int timeout_ms, int64_t begin)
{/*{{{*/
    struct timeval tv;
    gettimeofday(&tv, NULL);
    int used=tv.tv_sec*1000LL+tv.tv_usec/1000-begin;
    return(timeout_ms > used ? timeout_ms-used : 0);
}/*}}}*/

struct _unit
{/*{{{*/
    char from[NAME_SIZE+1];
    unsigned int syn;
    unsigned char buf[MSGQ_MSGSIZE_MAX];
};/*}}}*/
static int UNIT_MAX_SIZE=sizeof(struct _unit);
#define UNIT_HEAD_SIZE      (UNIT_MAX_SIZE-MSGQ_MSGSIZE_MAX)
#define UNIT_SIZE(msize)    (UNIT_HEAD_SIZE+msize)

struct _queue
{/*{{{*/
    char name[NAME_SIZE+1];
    mqd_t fd;
};/*}}}*/

struct _msgq_send
{/*{{{*/
    struct _unit unit;
    struct _queue queues[MSGQ_QUEUES_MAX];
    int upper;
};/*}}}*/

int msgq_send_attach(MSGQ_SEND *p_msgq)
{/*{{{*/
    if(!p_msgq) return(RET_ERR_PARA);

    MSGQ_SEND msgq;
    ALLOC_RETERR(msgq, struct _msgq_send, RET_ERR_ALLOC);
    int i;
    for(i=0; i<MSGQ_QUEUES_MAX; ++i)
    {
        CLR_BUF(msgq->queues[i].name);
    }
    msgq->upper=0;

    *p_msgq=msgq;
    return(0);
}/*}}}*/

int msgq_send_detach(MSGQ_SEND msgq)
{/*{{{*/
    if(!msgq) return(RET_ERR_PARA);

    int i;
    for(i=0; i<msgq->upper; ++i)
    {
        struct _queue *q=&msgq->queues[i];
        if(!BUF_IS_CLR(q->name))
            mq_close(q->fd);
    }
    free(msgq);

    return(0);
}/*}}}*/

int msgq_send(MSGQ_SEND msgq, char *from, char *to,
        unsigned int syn, unsigned char *msg, int msg_size,
        int timeout_ms)
{/*{{{*/
    if(!msgq || !to ||
            !msg || msg_size < 0 || msg_size > MSGQ_MSGSIZE_MAX)
        return(RET_ERR_PARA);
    char sqname[NAME_SIZE+1];
    strcpy(sqname, to);
    trim_space(sqname);
    if(BUF_IS_CLR(sqname)) return(RET_ERR_PARA);

    struct _unit *unit=&msgq->unit;
    if(from)
        strcpy(unit->from, from);
    else
        CLR_BUF(unit->from);
    unit->syn=syn;
    memcpy(unit->buf, msg, msg_size);

    struct timespec ts;
    int res;

    // 查询消息队列
    int i;
    int idle=-1;
    struct _queue *queue;
    for(i=0; i<msgq->upper; ++i)
    {
        queue=&msgq->queues[i];
        if(BUF_IS_CLR(queue->name))
        {
            if(idle == -1) idle=i;
            continue;
        }
        if(strcmp(queue->name, sqname) == 0)
        {
            // 发送
            if(timeout_ms >= 0)
            {
                _begin(timeout_ms, &ts);
                res=mq_timedsend(queue->fd, (char *)unit,
                        UNIT_SIZE(msg_size), 0, &ts);
                if(res == -1)
                {
                    int error=get_last_error(NULL);
                    if(error == ETIMEDOUT) return(RET_ERR_TIMEOUT);
                }
            }
            else
            {
                res=mq_send(queue->fd, (char *)unit,
                        UNIT_SIZE(msg_size), 0);
            }
            if(res == 0) return(0);
            CLR_BUF(queue->name);
            mq_close(queue->fd);
            if(idle == -1) idle=i; 
            break;
        }
    }
    if(idle == -1)
    {
        if(msgq->upper == MSGQ_QUEUES_MAX) return(-1);
        idle=msgq->upper;
        ++msgq->upper;
    }
    queue=&msgq->queues[idle];

    // 新连接
    char sq[NAME_SIZE+1];
    sprintf(sq, "/%s", sqname);
    int oflag=O_RDWR;
    mqd_t fd=mq_open(sq, oflag);
    if(fd == -1) return(-2);
    strcpy(queue->name, sqname);
    queue->fd=fd;

    // 发送
    if(timeout_ms >= 0)
    {
        _begin(timeout_ms, &ts);
        res=mq_timedsend(queue->fd, (char *)unit, UNIT_SIZE(msg_size),
                0, &ts);
        if(res == -1)
        {
            int error=get_last_error(NULL);
            if(error == ETIMEDOUT) return(RET_ERR_TIMEOUT);
            CLR_BUF(queue->name);
            mq_close(queue->fd);
            return(-3);
        }
    }
    else
    {
        res=mq_send(queue->fd, (char *)unit, UNIT_SIZE(msg_size), 0);
        if(res == -1)
        {
            CLR_BUF(queue->name);
            mq_close(queue->fd);
            return(-3);
        }
    }

    return(0);
}/*}}}*/

struct _msgq_recv
{/*{{{*/
    struct _unit unit;
    struct _queue queue;
};/*}}}*/

int msgq_recv_attach(char *queue_name, MSGQ_RECV *p_msgq)
{/*{{{*/
    if(!p_msgq) return(RET_ERR_PARA);
    *p_msgq=NULL;
    if(!queue_name|| strlen(queue_name) > NAME_SIZE)
        return(RET_ERR_PARA);
    char qname[NAME_SIZE+1];
    strcpy(qname, queue_name);
    trim_space(qname);
    if(BUF_IS_CLR(qname)) return(RET_ERR_PARA);

    MSGQ_RECV msgq;
    ALLOC_RETERR(msgq, struct _msgq_recv, RET_ERR_ALLOC);

    char queue[NAME_SIZE+1];
    sprintf(queue, "/%s", qname);
    int oflag=(O_CREAT | O_RDWR);
    int mode=(S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP);
    struct mq_attr attr;
    attr.mq_flags=0;
    attr.mq_maxmsg=MSGQ_MSG_MAX;
    attr.mq_msgsize=UNIT_MAX_SIZE;
    attr.mq_curmsgs=0;
    mqd_t fd=mq_open(queue, oflag, mode, &attr);
    if(fd == -1)
    {
        free(msgq); return(-1);
    }
    strcpy(msgq->queue.name, qname);
    msgq->queue.fd=fd;

    *p_msgq=msgq;
    return(0);
}/*}}}*/

int msgq_recv_detach(MSGQ_RECV msgq)
{/*{{{*/
    if(!msgq) return(RET_ERR_PARA);
    mq_close(msgq->queue.fd);
    free(msgq);
    return(0);
}/*}}}*/

int msgq_recv(MSGQ_RECV msgq, char *from, 
        unsigned int *syn, unsigned char *msg, int *msg_size,
        int timeout_ms)
{/*{{{*/
    if(!msgq || !msg || !msg_size)
        return(RET_ERR_PARA);

    struct _unit *unit=&msgq->unit;

    int len;
    if(timeout_ms >= 0)
    {
        struct timespec ts;
        _begin(timeout_ms, &ts);
        len=mq_timedreceive(msgq->queue.fd,
                (char *)unit, UNIT_MAX_SIZE,
                NULL, &ts);
        if(len == -1)
        {
            int error=get_last_error(NULL);
            if(error == ETIMEDOUT) return(RET_ERR_TIMEOUT);
            return(-1);
        }
    }
    else
    {
        len=mq_receive(msgq->queue.fd,
                (char *)unit, UNIT_MAX_SIZE, NULL);
        if(len == -1) return(-1);
    }
    if(len < UNIT_HEAD_SIZE) return(-2);
    len-=UNIT_HEAD_SIZE;

    if(from) strcpy(from, unit->from);
    if(syn) *syn=unit->syn;
    memcpy(msg, unit->buf, len);
    *msg_size=len;
    return(0);
}/*}}}*/

int msgq_clear(char *queue_name)
{/*{{{*/
    if(!queue_name|| strlen(queue_name) > NAME_SIZE)
        return(RET_ERR_PARA);
    char qname[NAME_SIZE+1];
    strcpy(qname, queue_name);
    trim_space(qname);
    if(BUF_IS_CLR(qname)) return(RET_ERR_PARA);

    char queue[FILE_NM_SIZE+1];
    sprintf(queue, "/%s", qname);
    int oflag=(O_CREAT | O_RDWR | O_NONBLOCK);
    int mode=(S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP);
    struct mq_attr attr;
    attr.mq_flags=0;
    attr.mq_maxmsg=MSGQ_MSG_MAX;
    attr.mq_msgsize=UNIT_MAX_SIZE;
    attr.mq_curmsgs=0;
    mqd_t fd=mq_open(queue, oflag, mode, &attr);
    if(fd == -1) return(-1);

    int rtn;
    char *buf;
    ALLOC_MULTI_DOERR(buf, char, attr.mq_msgsize,
            rtn=-2; goto GO_OUT;
            );
    int res;
    while(1)
    {
        res=mq_receive(fd, buf, attr.mq_msgsize, NULL);
        if(res == -1)
        {
            if(get_last_error(NULL) == EAGAIN)
                rtn=0;
            else
                rtn=-3;
            break;
        }
    }

GO_OUT:
    if(buf) free(buf);
    mq_close(fd);
    return(rtn);
}/*}}}*/

struct _msgq_syn
{/*{{{*/
    struct _unit unit;
    unsigned int syn;
    struct
    {
        struct _queue queues[MSGQ_QUEUES_MAX];
        int upper;
    } _send_;
    struct
    {
        struct _queue queue;
    } _recv_;
};/*}}}*/

int msgq_syn_attach(char *recv_queue_name, MSGQ_SYN *msgqp)
{/*{{{*/
    if(!msgqp)
        return(RET_ERR_PARA);
    *msgqp=NULL;
    if(!recv_queue_name|| strlen(recv_queue_name) > NAME_SIZE)
        return(RET_ERR_PARA);
    char rqname[NAME_SIZE+1];
    strcpy(rqname, recv_queue_name);
    trim_space(rqname);
    if(BUF_IS_CLR(rqname)) return(RET_ERR_PARA);

    MSGQ_SYN msgq;
    ALLOC_RETERR(msgq, struct _msgq_syn, RET_ERR_ALLOC);
    msgq->syn=0;
    int i;
    for(i=0; i<MSGQ_QUEUES_MAX; ++i)
    {
        CLR_BUF(msgq->_send_.queues[i].name);
    }
    msgq->_send_.upper=0;

    int rtn;

    char rq[NAME_SIZE+1];
    sprintf(rq, "/%s", rqname);
    //if(mq_unlink(rq))
    //{
    //    int error=get_last_error(NULL);
    //    if(error != ENOENT)
    //    {
    //        rtn=-1; goto GO_OUT;
    //    }
    //}

    int oflag=(O_CREAT | O_RDWR);
    int mode=(S_IRUSR|S_IWUSR|S_IRGRP|S_IWGRP);
    struct mq_attr attr;
    attr.mq_flags=0;
    attr.mq_maxmsg=MSGQ_MSG_MAX;
    attr.mq_msgsize=UNIT_MAX_SIZE;
    attr.mq_curmsgs=0;
    mqd_t fd=mq_open(rq, oflag, mode, &attr);
    if(fd == -1)
    {
        rtn=-2; goto GO_OUT;
    }
    strcpy(msgq->_recv_.queue.name, recv_queue_name);
    msgq->_recv_.queue.fd=fd;

    *msgqp=msgq;
    return(0);

GO_OUT:
    free(msgq);
    return(rtn);
}/*}}}*/

int msgq_syn_detach(MSGQ_SYN msgq)
{/*{{{*/
    if(!msgq)
        return(RET_ERR_PARA);
    int i;
    for(i=0; i<msgq->_send_.upper; ++i)
    {
        struct _queue *q=&msgq->_send_.queues[i];
        if(!BUF_IS_CLR(q->name))
            mq_close(q->fd);
    }
    mq_close(msgq->_recv_.queue.fd);
    //char rq[NAME_SIZE+1];
    //sprintf(rq, "/%s", msgq->_recv_.queue.name);
    //mq_unlink(rq);

    free(msgq);
    return(0);
}/*}}}*/

static int _syn_send(struct _msgq_syn *msgq,
        char *sqname, unsigned int syn,
        unsigned char *msg, int msg_size, int *o_timeout_ms)
{/*{{{*/
    struct _unit *unit=&msgq->unit;
    strcpy(unit->from, msgq->_recv_.queue.name);
    unit->syn=syn;
    memcpy(unit->buf, msg, msg_size);

    int timeout_ms=*o_timeout_ms;
    struct timespec ts;
    int64_t begin;
    int res;

    // 查询消息队列
    int i;
    int idle=-1;
    struct _queue *queue;
    for(i=0; i<msgq->_send_.upper; ++i)
    {
        queue=&msgq->_send_.queues[i];
        if(BUF_IS_CLR(queue->name))
        {
            if(idle == -1) idle=i;
            continue;
        }
        if(strcmp(queue->name, sqname) == 0)
        {
            // 发送
            timeout_ms=*o_timeout_ms;
            if(timeout_ms >= 0)
            {
                begin=_begin(timeout_ms, &ts);
                res=mq_timedsend(queue->fd, (char *)unit,
                        UNIT_SIZE(msg_size), 0, &ts);
                if(res == -1)
                {
                    int error=get_last_error(NULL);
                    if(error == ETIMEDOUT) return(RET_ERR_TIMEOUT);
                }
                *o_timeout_ms=_left(timeout_ms, begin);
            }
            else
            {
                res=mq_send(queue->fd, (char *)unit,
                        UNIT_SIZE(msg_size), 0);
            }
            if(res == 0) return(0);
            //fprintf(stderr, "queue(%s) send fail, reopen\n", queue->name);
            CLR_BUF(queue->name);
            mq_close(queue->fd);
            if(idle == -1) idle=i; 
            break;
        }
    }
    if(idle == -1)
    {
        if(msgq->_send_.upper == MSGQ_QUEUES_MAX)
            return(-1);
        idle=msgq->_send_.upper;
        ++msgq->_send_.upper;
    }
    queue=&msgq->_send_.queues[idle];

    // 新连接
    char sq[NAME_SIZE+1];
    sprintf(sq, "/%s", sqname);
    int oflag=O_RDWR;
    mqd_t fd=mq_open(sq, oflag);
    if(fd == -1) return(-2);
    strcpy(queue->name, sqname);
    queue->fd=fd;

    // 发送
    timeout_ms=*o_timeout_ms;
    if(timeout_ms >= 0)
    {
        begin=_begin(timeout_ms, &ts);
        res=mq_timedsend(queue->fd, (char *)unit, UNIT_SIZE(msg_size),
                0, &ts);
        if(res == -1)
        {
            int error=get_last_error(NULL);
            if(error == ETIMEDOUT) return(RET_ERR_TIMEOUT);
            CLR_BUF(queue->name);
            mq_close(queue->fd);
            return(-3);
        }
        *o_timeout_ms=_left(timeout_ms, begin);
    }
    else
    {
        res=mq_send(queue->fd, (char *)unit, UNIT_SIZE(msg_size), 0);
        if(res == -1)
        {
            CLR_BUF(queue->name);
            mq_close(queue->fd);
            return(-3);
        }
    }

    return(0);
}/*}}}*/

static int _syn_recv(struct _msgq_syn *msgq,
        unsigned char *msg, int *msg_size, int *o_timeout_ms)
{/*{{{*/
    struct _unit *unit=&msgq->unit;

    int timeout_ms=*o_timeout_ms;
    struct timespec ts;
    int64_t begin;
    int len;

    if(timeout_ms >= 0)
    {
        begin=_begin(timeout_ms, &ts);
        len=mq_timedreceive(msgq->_recv_.queue.fd,
                (char *)unit, UNIT_MAX_SIZE,
                NULL, &ts);
        if(len == -1)
        {
            int error=get_last_error(NULL);
            if(error == ETIMEDOUT) return(RET_ERR_TIMEOUT);
            fprintf(stderr, "recv fail, error:%d\n", error);
            return(-1);
        }
        *o_timeout_ms=_left(timeout_ms, begin);
    }
    else
    {
        len=mq_receive(msgq->_recv_.queue.fd,
                (char *)unit, UNIT_MAX_SIZE, NULL);
        if(len == -1) return(-1);
    }
    if(len < UNIT_HEAD_SIZE) return(-2);
    len-=UNIT_HEAD_SIZE;
    //if(len != unit->msg_size)
    //{
    //    return(-3);
    //}

    memcpy(msg, unit->buf, len);
    *msg_size=len;
    return(0);
}/*}}}*/

int msgq_syn_call(MSGQ_SYN msgq, char *send_queue_name,
        unsigned char *request, int request_size,
        unsigned char *respond, int *respond_size,
        int timeout_ms)
{/*{{{*/
    if(!msgq ||
            !send_queue_name || strlen(send_queue_name) > NAME_SIZE ||
            !request || request_size <= 0 ||
            !respond || !respond_size)
        return(RET_ERR_PARA);
    char sqname[NAME_SIZE+1];
    strcpy(sqname, send_queue_name);
    trim_space(sqname);
    if(BUF_IS_CLR(sqname))
        return(RET_ERR_PARA);

    int res;
    res=_syn_send(msgq, sqname, msgq->syn, request, request_size,
            &timeout_ms);
    if(res)
    {
        return(-100+res);
    }
    do
    {
        res=_syn_recv(msgq, respond, respond_size, &timeout_ms);
        if(res)
            return(-200+res);
    } while(msgq->syn != msgq->unit.syn);
    ++msgq->syn;

    return(0);
}/*}}}*/

int msgq_syn_fetch(MSGQ_SYN msgq,
        unsigned char *msg, int *msg_size,
        int timeout_ms)
{/*{{{*/
    if(!msgq || !msg || !msg_size) return(RET_ERR_PARA);

    int res=_syn_recv(msgq, msg, msg_size, &timeout_ms);
    return(res);
}/*}}}*/

int msgq_syn_reply(MSGQ_SYN msgq,
        unsigned char *msg, int msg_size,
        int timeout_ms)
{/*{{{*/
    if(!msgq || !msg || msg_size<0) return(RET_ERR_PARA);

    char sqname[NAME_SIZE+1];
    strcpy(sqname, msgq->unit.from);
    int res=_syn_send(msgq, sqname, msgq->unit.syn,
            msg, msg_size, &timeout_ms);
    return(res);
}/*}}}*/
